package com.yuan.subscriber;

import com.alibaba.fastjson.JSONObject;
import com.yuan.constant.GlobalConstant;
import com.yuan.entity.dao.ChatMsg;
import com.yuan.entity.resp.ChatResultVo;
import com.yuan.service.ChatMsgService;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.FluxSink;

import java.util.Date;


@Slf4j
public class DsSubscriber implements Subscriber<String>, Disposable {


    private final FluxSink<String> emitter;

    private Subscription subscription;
    private final StringBuilder sb;
    private final CompletedCallBack completedCallBack;
    private String sessionId;
    private ChatMsgService chatMsgService;
    private String msgId ;

    public DsSubscriber(FluxSink<String> emitter, CompletedCallBack completedCallBack, String sessionId , ChatMsgService chatMsgService ) {
        this.emitter = emitter;
        this.sb = new StringBuilder();
        this.completedCallBack = completedCallBack;
        this.sessionId = sessionId;
        this.chatMsgService = chatMsgService;
        this.msgId = "";
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(String data) {
        log.info("Deepseek 返回数据：{}", data);
        subscription.request(1);

        ChatResultVo vo = JSONObject.parseObject(data, ChatResultVo.class);
        sb.append(vo.getMessage());
        msgId = vo.getMessageId();

        emitter.next(vo.getMessage());
    }

    @Override
    public void onError(Throwable throwable) {
        log.error("Deepseek 返回数据异常：{}", throwable.getMessage());
        emitter.error(throwable);
        completedCallBack.fail("");
    }

    @Override
    public void onComplete() {
        log.info("Deepseek 返回数据完成");
        // save to db


        ChatMsg chatMsg = new ChatMsg();
        chatMsg.setContent(sb.toString());
        chatMsg.setRole(GlobalConstant.ROLE.ASSISTANT);
        chatMsg.setCreateTime(new Date());
        //chatMsg.setDeepThinking(chatParam.isDeepThinking());
        //chatMsg.setOnlineSearch(chatParam.isOnlineSearch());
        chatMsg.setSessionId(sessionId);
        //chatMsg.setBotId(botId);
        chatMsg.setMsgId(msgId);
        chatMsgService.insertByEntity(chatMsg);

        completedCallBack.completed( sb.toString());
        emitter.complete();
    }

    @Override
    public void dispose() {
        log.info("dispose");
    }
}
